package com.csx.midware.redis;

import redis.clients.jedis.Jedis;


/**
 * RedisQueueExample 类展示了如何使用 Redis 实现简单的队列功能。
 * 实际应用包括：
 * - 异步任务处理：将耗时任务放入队列，由后台工作进程异步处理。
 * - 消息传递：在分布式系统中，不同服务之间通过队列进行消息传递。
 * - 流量削峰：处理高并发请求时，使用队列缓冲请求，避免直接冲击数据库。
 * - 缓存更新：将缓存更新操作放入队列，确保顺序执行，防止并发更新导致的数据不一致。
 * - 日志收集：将日志先写入队列，再由专门的日志处理程序批量读取并保存到持久化存储中。
 */
public class RedisQueueExample {

    private Jedis jedis;

    public RedisQueueExample() {
        jedis = new Jedis("localhost");
    }

    public void enqueue(String queueName, String message) {
        jedis.rpush(queueName, message);
    }

    public String dequeue(String queueName) {
        return jedis.lpop(queueName);
    }

    public void acknowledge(String queueName, String message) {
        // 确认消息已处理，可以在这里执行其他确认操作
    }

    public void requeue(String queueName, String message) {
        jedis.rpush(queueName, message);
    }

    public void handleFailedMessage(String queueName, String message) {
        String dlqName = queueName + ":dlq";
        jedis.rpush(dlqName, message);
    }

    public void close() {
        jedis.close();
    }

    public static void main(String[] args) {
        Producer producer = new Producer();
        String queueName = "myQueue";
        Consumer consumer = new Consumer();

        producer.produce(queueName, "Message 1");
        producer.produce(queueName, "Message 2");
        producer.produce(queueName, "Message 3");



        consumer.consume(queueName);

        consumer.close();
        producer.close();
    }


    static class Producer {

        private RedisQueueExample redisQueue;

        public Producer() {
            redisQueue = new RedisQueueExample();
        }

        public void produce(String queueName, String message) {
            redisQueue.enqueue(queueName, message);
            System.out.println("Produced: " + message);
        }

        public void close() {
            redisQueue.close();
        }


    }

    static class Consumer {

        private RedisQueueExample redisQueue;

        public Consumer() {
            redisQueue = new RedisQueueExample();
        }

        public void consume(String queueName) {
            String message;
            while ((message = redisQueue.dequeue(queueName)) != null) {
                try {
                    processMessage(message);
                    redisQueue.acknowledge(queueName, message);
                    System.out.println("consume: " + message);
                } catch (Exception e) {
                    System.err.println("Failed to process message: " + message + e);
                    redisQueue.handleFailedMessage(queueName, message);
                }
            }
        }

        private void processMessage(String message) throws Exception {
            // 模拟消息处理逻辑
            if (message.equals("Message 2")) {
                throw new Exception("Simulated processing failure");
            }
            // 正常处理逻辑
        }

        public void close() {
            redisQueue.close();
        }
    }

}
